package com.rest.util;

import com.rest.exception.ApiException;
import com.rest.response.BaseResponse;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Function;

import static io.reactivex.schedulers.Schedulers.io;

/**
 * @Description: java类作用描述
 * @Author: lzp
 * @CreateDate: 2022/2/23 15:52
 * @UpdateRemark: 更新说明
 */
public class RxUtil {


    /**
     * 统一线程处理
     *
     */
    public static <T> ObservableTransformer<T, T> applySchedulers() {
        return upstream -> upstream.subscribeOn(io())
                .unsubscribeOn(io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 对服务器返回数据进行预处理
     *
     * @param <T>
     * @return
     */
    public static <T> ObservableTransformer<BaseResponse<T>, T> flatResponse() {
        return upstream -> upstream.flatMap((Function<BaseResponse<T>, ObservableSource<T>>) result -> {
            if (result.isSuccessful()) {
                return createData(result.data);
            } else {
                return Observable.error(new ApiException(result.code, result.message));
            }
        });
    }

    public static <T> Observable<T> createData(final T data) {
        return Observable.create(emitter -> {
            try {
                emitter.onNext(data);
                emitter.onComplete();
            } catch (Exception e) {
                emitter.onError(e);
            }
        });
    }


    public static <T> FlowableTransformer<T, T> rxSchedulerHelper() {
        return observable -> observable.subscribeOn(io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 生成Flowable
     *
     * @param <T>
     * @return
     */
    public static <T> Flowable<T> createFlowable(final T data) {
        return Flowable.create(emitter -> {
            try {
                emitter.onNext(data);
                emitter.onComplete();
            } catch (Exception e) {
                emitter.onError(e);
            }
        }, BackpressureStrategy.BUFFER);
    }
}
